package com.mjf.networkflow_analysis

import com.mjf.dim.{UserBehavior, UvCount}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * UV 简单实现
 */
object UniqueVisitor {
  def main(args: Array[String]): Unit = {

    // 创建流处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 定义时间语义
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 从文件读取数据
    val inputStream: DataStream[String] = env.readTextFile("D:\\coding\\idea\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv")

    // 将数据转换为样例类，并且提取timestamp定义watermark
    val dataStream: DataStream[UserBehavior] = inputStream.map {
      line =>
        val dataArray: Array[String] = line.split(",")
        UserBehavior(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong)
    }.assignAscendingTimestamps(_.timestamp * 1000L)
    
    // 分配key,包装成二元组开窗聚合
    val uvStream: DataStream[UvCount] = dataStream
      .filter(_.behavior == "pv")
      .timeWindowAll(Time.hours(1)) // 基于DataStream开一小时的滚动窗口进行统计
//      .apply(new UvCountResult()) // 全窗口函数实现
      .aggregate(new UvCountAgg(), new UvCountResultWithIncrAgg())  // 增量聚合函数实现

    uvStream.print()

    env.execute("uv job")

  }
}

// 自定义全窗口函数
class UvCountResult() extends AllWindowFunction[UserBehavior, UvCount, TimeWindow] {
  override def apply(window: TimeWindow, input: Iterable[UserBehavior], out: Collector[UvCount]): Unit = {
    // 定义一个Set类型,来保存所有的userId,自动去重
    var idSet: Set[Long] = Set[Long]()

    // 将当前窗口的所有数据，添加到set
    for (userBehavior <- input) {
      idSet += userBehavior.userId
    }

    // 输出set的大小，就是去重之后的UV值
    out.collect(UvCount(window.getEnd, idSet.size.toLong))
  }
}

// 自定义增量聚合函数,需要定义一个Set作为累加状态
class UvCountAgg() extends AggregateFunction[UserBehavior, Set[Long], Long] {
  override def createAccumulator(): Set[Long] = Set[Long]()

  override def add(value: UserBehavior, accumulator: Set[Long]): Set[Long] = accumulator + value.userId

  override def getResult(accumulator: Set[Long]): Long = accumulator.size.toLong

  override def merge(a: Set[Long], b: Set[Long]): Set[Long] = a ++ b
}

// 自定义窗口函数，添加Window信息，包装成样例类
class UvCountResultWithIncrAgg() extends AllWindowFunction[Long, UvCount, TimeWindow] {
  override def apply(window: TimeWindow, input: Iterable[Long], out: Collector[UvCount]): Unit = {
    out.collect(UvCount(window.getEnd, input.head))
  }
}










